package cn.doitedu.dmp

import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

object GraphxDemo {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("图计算最大连通子图算法示例")
      .master("local")
      .getOrCreate()

    import spark.implicits._

    // 加载数据
    val ds: Dataset[String] = spark.read.textFile("user_portrait/data/graphx/input/demo.txt")


    // 提取出数据中的标识
    val ids = ds.rdd.map(line=>{
      val split = line.split(",")
      Array(split(1),split(2))
    })

    // 转换成Graphx API中的 Vertex（点）集合
    // Vertex实质上就是一个元组(Long 点id, [T] 属性数据)
    val vertexArrayRDD: RDD[Array[(Long, String)]] = ids.map(arr=>{
      for (userFlag <- arr) yield (userFlag.hashCode.toLong,userFlag)
    })

    // 打散成点集合
    val vertexRdd: RDD[(Long, String)] = vertexArrayRDD.flatMap(arr => arr).distinct()


    // 根据数据生成 边集合
    // graphx中的边的封装类型为 Edge
    val edgeRdd: RDD[Edge[String]] = vertexArrayRDD.flatMap(arr=>{
      val firstVertex = arr(0)
      for (i <- 1 until arr.length) yield Edge(firstVertex._1,arr(i)._1,firstVertex._2+"->"+arr(i)._2)
    })


    // 将数据中的所有点的集合，所有边的集合，构造一个图Graph对象
    val graph = Graph(vertexRdd, edgeRdd)

    // 在图上调用“最大连通子图算法”即可得到连通子图结果
    val resGraph = graph.connectedComponents()


    // 取出结果图中的 点集合，即可看出哪些点属于同一个连通子图
    val resVertices: RDD[(Long,Long)] = resGraph.vertices


    // 打印结果：结果图中的每一个点，都带上了“连通子图标识属性”，属于同一个连通子图的点，属性相同
    resVertices.foreach(println)


    // 从点ID结果，还原到业务数据的结果
    val res = vertexRdd.join(resVertices).map(tp=>tp._2)

    res.toDF("userFlag","connectedId").show(100,false)

    res.toDF("userFlag","connectedId").createTempView("x")
    spark.sql(
      """
        |select
        |connectedId,collect_set(userFlag) as flags
        |from x
        |group by connectedId
        |
        |
        |""".stripMargin)
        .show(100,false)

    spark.close()
  }

}
